#!/usr/bin/python3
# -*- coding: utf-8 -*-

import json
import os
import platform
import shutil
import subprocess
import sys
from collections import namedtuple
from typing import Any, Dict, List, Optional

import psutil
import requests
from requests import request

from databases.sqlite import SQLiteDB
from databases.table import create_tables, delete_db
from public import description
from public.project_patch import (ALLURE_JSON, ALLURE_REPORT,
                                  ALLURE_SUMMARY_FILE, CONFIG_DIR, CONFIG_YAML,
                                  DB, NOTIFICATION_YAML, REPORT_DIR)
from utils.cache_util import cache_manager
from utils.date_util import format_timer
from utils.file_util import zip_folder
from utils.log import log
from utils.yaml_util import YamlHandler


class SetupSession:

    @staticmethod
    def write_config():
        """
        写env
        :return:
        """
        test_config = YamlHandler(CONFIG_YAML).read_yaml()
        env = os.environ['TEST_ENV']
        test_config["env"] = env
        YamlHandler(CONFIG_YAML).write_yaml(test_config)

    @staticmethod
    def load_yaml_config_to_cache() -> None:
        """
        将配置文件写入缓存
        :return:
        """
        env = os.environ['TEST_ENV']
        filename = f"{env}_config.yaml"
        config_file = CONFIG_DIR / filename
        if not config_file.exists():
            log.warning(f"警告: 文件 {config_file} 不存在,请检查配置文件")
            sys.exit(1)
        log.info(f"正在加载{env} 环境的配置文件: {config_file}")
        config = YamlHandler(config_file).read_yaml()
        cache_manager.set('env', env)
        for key, value in config.items():
            cache_manager.set(key, value)

    @staticmethod
    def setup_echo() -> None:
        """
        打印环境信息
        :return:
        """
        log.info(description)
        env = os.environ['TEST_ENV']
        log.info(f"测试环境:{env}")

    @staticmethod
    def setup_db_table() -> None:
        """
        创建测试会话期间的db
        :return:
        """
        delete_db(DB)
        create_tables(DB)

    @staticmethod
    def write_env_to_db() -> None:
        """
        写测试环境到数据表
        :return:
        """
        sqlite = SQLiteDB(DB)
        base_data = {
            "env": cache_manager.get("env"),
            "region": cache_manager.get("region"),
            "zone": cache_manager.get("zone"),
            "login_url": cache_manager.get("login_url"),
            "base_url": cache_manager.get("base_url"),
        }
        for key, value in base_data.items():
            value = json.dumps(value)
            sqlite.insert_data(table="config",
                               data={
                                   "key": key,
                                   "value": value},
                               )

    @staticmethod
    def write_test_config_to_db() -> None:
        """
        写测试环境到数据表。
        :return:
        """
        sqlite = SQLiteDB(DB)
        datas = cache_manager.get_all()
        for key, value in datas.items():
            value = json.dumps(value)
            sqlite.insert_data(table="config", data={"key": key, "value": value})

    @staticmethod
    def delete_allure_json() -> None:
        """
        删除上一次测试报告数据
        :return:
        """
        if ALLURE_JSON.exists():
            for item in ALLURE_JSON.iterdir():
                item_path = ALLURE_JSON / item
                if item_path.is_file():
                    item_path.unlink(missing_ok=True)

    @staticmethod
    def create_init_dir() -> None:
        """
        创建初始化文件夹
        :return:
        """
        if not ALLURE_REPORT.exists():
            ALLURE_REPORT.mkdir(parents=True, exist_ok=True)
        if not ALLURE_JSON.exists():
            ALLURE_JSON.mkdir(parents=True, exist_ok=True)


class TeardownSession:
    @staticmethod
    def mk_trend() -> None:
        """
        将上次测试报告中的测试历史数据拷贝到最新的生成的历史数据中
        @param config: 项目配置类
        @return:
        """
        allure_report_history_path = ALLURE_REPORT / "history"
        allure_results_history_path = ALLURE_JSON / "history"
        if allure_results_history_path.exists():
            shutil.rmtree(allure_results_history_path)
        if allure_report_history_path.exists():
            shutil.copytree(allure_report_history_path, allure_results_history_path)

    @staticmethod
    def write_env_to_allure() -> None:
        """
        写入环境信息到allure报告
        :return:
        """
        content = ""
        base_data = {
            "env": cache_manager.get("env"),
            "region": cache_manager.get("region"),
            "zone": cache_manager.get("zone"),
            "login_url": cache_manager.get("login_url"),
            "base_url": cache_manager.get("base_url"),
        }
        for k, v in base_data.items():
            content += f"{k}={v}\n"
        sys_info = SystemInfo().get_system_info()
        for k, v in sys_info.items():
            content += f"{k}={v}\n"
        with open(
                ALLURE_JSON / "environment.properties", mode="w", encoding="utf-8"
        ) as f:
            f.write(content)

    @staticmethod
    def write_categories_to_allure() -> None:
        """
        写入测试用例结果的分类到文件categories.json中
        默认情况下，有两类缺陷：
        Product defects 产品缺陷（测试结果：failed）
        Test defects 测试缺陷（测试结果：error/broken）
        :return:
        """
        categories_json = [
            {"name": "忽略执行的用例", "matchedStatuses": ["skipped"]},
            {"name": "执行失败的用例", "matchedStatuses": ["failed"]},
            {"name": "测试用例错误", "matchedStatuses": ["broken"]},
            {"name": "执行成功的用例", "matchedStatuses": ["passed"]},
            {"name": "未知状态的用例", "matchedStatuses": ["unknown"]},
        ]
        with open(ALLURE_JSON / "categories.json", mode="w", encoding="utf-8") as f:
            f.write(json.dumps(categories_json, indent=2, ensure_ascii=False))

    @staticmethod
    def generate_allure_report() -> None:
        """
        生成测试报告
        :return:
        """
        try:
            if not ALLURE_JSON.exists():
                ALLURE_JSON.mkdir(parents=True, exist_ok=True)
            if not ALLURE_REPORT.exists():
                ALLURE_REPORT.mkdir(parents=True, exist_ok=True)
            cmd = f"allure generate {ALLURE_JSON} -o {ALLURE_REPORT} -c"
            log.info(f"生成测试报告: {cmd}")
            subprocess.run(cmd.split(), check=True)
        except Exception as e:
            log.error(f"生成测试报告时发生错误: {e}")

    @staticmethod
    def send_report_to_allure_server() -> None or Dict:
        """
        发送测试报告到allure-server
        :return:
        """
        allure_server_conf = YamlHandler(NOTIFICATION_YAML).read_yaml()["allure_server"]
        url = allure_server_conf.get("url")
        port = allure_server_conf.get("port")
        if not all([url, port]):
            log.error("allure-server url or port未配置，请检查配置文件")
            return
        filepath = zip_folder(ALLURE_JSON, REPORT_DIR / "allure-report")
        if filepath and filepath.exists():
            allure_server_connect = AllureServerConnection(filepath=filepath,
                                                           url=url,
                                                           port=port)
            latest_report_link = allure_server_connect.send_report_to_allure_server()
            filepath.unlink(missing_ok=True)
            return latest_report_link


class AllureReportSummary:
    @staticmethod
    def get_summary(allure_summary_json):
        """
        获取所有 allure 报告中执行用例的情况
        @param allure_summary_json: 文件地址,位于allure report/widgets/summary.json
        @return:
        """
        Summary = namedtuple(
            "Summary",
            [
                "total",
                "passed",
                "broken",
                "failed",
                "skipped",
                "passed_rate",
                "failed_rate",
                "run_time",
                "unknown",
            ],
        )

        if os.path.exists(allure_summary_json):
            with open(allure_summary_json, "r", encoding="utf-8") as f:
                data = json.load(f)
            total = (
                data["statistic"]["total"]
                if data["statistic"]["total"] is not None
                else 0
            )
            passed = (
                data["statistic"]["passed"]
                if data["statistic"]["passed"] is not None
                else 0
            )
            broken = (
                data["statistic"]["broken"]
                if data["statistic"]["broken"] is not None
                else 0
            )
            failed = (
                data["statistic"]["failed"]
                if data["statistic"]["failed"] is not None
                else 0
            )
            skipped = (
                data["statistic"]["skipped"]
                if data["statistic"]["skipped"] is not None
                else 0
            )
            unknown = (
                data["statistic"]["unknown"]
                if data["statistic"]["unknown"] is not None
                else 0
            )
            try:
                passed_rate = round(passed / (total - skipped))
                failed_rate = round(failed / (total - skipped))
                run_time = round(data["time"]["duration"] / 1000, 2)
            except ZeroDivisionError:
                passed_rate = 0
                failed_rate = 0
                run_time = 0

            summary = Summary(
                total=total,
                passed=passed,
                failed=failed,
                broken=broken,
                unknown=unknown,
                skipped=skipped,
                passed_rate=passed_rate,
                failed_rate=failed_rate,
                run_time=run_time,
            )
            return summary
        else:
            return {}


class AllureServerConnection:
    def __init__(self, filepath: str = None, url: str = None, port: int = 8080):
        self.filepath = filepath
        self.port = str(port)
        self.url = url + ":" + self.port

    def generate_allure_report_uuid(self, filepath: str) -> Dict[str, Any] or None:
        """
        获取uuid
        :param filepath:
        :return:
        """
        if not filepath:
            log.error("文件路径不能为空")
            return None
        try:
            with open(filepath, "rb") as file_data:
                files = {
                    "allureResults": (
                        os.path.basename(filepath),
                        file_data,
                        "application/zip",
                    )
                }
                response = requests.post(url=f"{self.url}/api/result", files=files)
                response.raise_for_status()  # 检查请求是否成功
                return response.json()
        except requests.RequestException as req_err:
            log.error(f"网络请求错误，生成uuid失败，错误信息：{req_err}")
        except FileNotFoundError as fnf_err:
            log.error(f"文件未找到，错误信息：{fnf_err}")
        except ValueError as val_err:  # JSON解析错误
            log.error(f"解析响应内容失败，错误信息：{val_err}")
        except Exception as e:
            log.error(f"未知错误，生成uuid失败，错误信息：{e}")
        return None

    def generate_allure_report(
            self, uuid: List[str] = None, build_name: str = None, branch: str = "master"
    ) -> Dict[str, Any] or None:
        """
        生成在线测试报告
        :build_name: 构建次数
        :branch: 构建分支,默认master
        :return:
        """
        if isinstance(uuid, str):
            uuid = [uuid]
        headers = {"Content-Type": "application/json"}
        data = {
            "reportSpec": {
                "path": [branch, build_name],
                "executorInfo": {"buildName": f"#{build_name}"},
            },
            "results": uuid,
            "deleteResults": False,
        }
        try:
            response = requests.post(
                f"{self.url}/api/report", headers=headers, data=json.dumps(data)
            )
            response.raise_for_status()
            return response.json()
        except requests.HTTPError as http_err:
            log.error(
                f"HTTP错误，生成测试报告失败，状态码：{http_err.response.status_code}, 错误信息：{http_err}"
            )
        except requests.RequestException as req_err:
            log.error(f"网络请求错误，生成测试报告失败，错误信息：{req_err}")
        except Exception as e:
            log.error(f"生成测试报告失败，未知错误信息：{e}")
        return None

    def send_report_to_allure_server(
            self, build_name: str = "1", branch: str = "master"
    ) -> Optional[Dict]:
        """
        发送测试报告到Allure服务器，并获取最新报告的链接。
        :param build_name: 构建编号，默认为'1'。
        :param branch: 分支名称，默认为'master'。
        :return: 包含最新报告链接的字典，若过程中发生错误则返回None。
        """
        try:
            uuid_response = self.generate_allure_report_uuid(self.filepath)
            if uuid_response is None:
                log.error("无法获取Allure报告的UUID")
                return None
            report_uuid = uuid_response.get("uuid")
            if not report_uuid:
                log.error("从响应中未获取到UUID")
                return None
            report_response = self.generate_allure_report(
                uuid=[report_uuid], build_name=build_name, branch=branch
            )
            if report_response is None:
                log.error("生成Allure测试报告失败")
                return None
            latest_report_link = report_response.get("latest")
            if latest_report_link:
                log.info(f"Allure测试报告生成成功，最新报告链接为: {latest_report_link}")
                return {"latest_report_link": latest_report_link}
            else:
                log.error("未能从报告响应中获取最新报告链接")
                return None
        except Exception as e:
            log.error(f"发送报告至Allure服务器过程中发生错误: {e}")
            return None


class SystemInfo:
    @staticmethod
    def get_system_info():
        """
        获取系统及平台信息
        :return:
        """
        system = platform.system()
        system_version = platform.version()
        node_name = platform.node()
        python_version = sys.version
        cpu_count = psutil.cpu_count(logical=True)

        mem_info = psutil.virtual_memory()
        total_memory = mem_info.total >> 20  # 转换为MB
        memory_percent = mem_info.percent

        disk_info = psutil.disk_usage("/")
        total_disk = disk_info.total >> 30  # 转换为GB
        disk_percent = disk_info.percent

        system_info_dict = {
            "system": system,
            "system_version": system_version,
            "node_name": node_name,
            "python_version": python_version,
            "cpu_count": cpu_count,
            "total_memory": total_memory,
            "memory_percent": memory_percent,
            "total_disk": total_disk,
            "disk_percent": disk_percent,
        }
        return system_info_dict


class WeChatSend:
    def __init__(self, tester="质量保障部门", title="自动化测试通知", report_address=""):
        self.wechat_conf = YamlHandler(NOTIFICATION_YAML).read_yaml()["wechat"]
        self.curl = self.wechat_conf["webhook"]
        if not self.curl:
            sys.exit("webhook 未配置，请检查！")
        self.headers = {"Content-Type": "application/json"}

        self.allure_report = AllureReportSummary.get_summary(ALLURE_SUMMARY_FILE)
        if not self.allure_report:
            sys.exit("allure summary文件不存在，请检查！")

        self.total = self.allure_report.total
        self.passed = self.allure_report.passed
        self.failed = self.allure_report.failed
        self.skipped = self.allure_report.skipped
        self.broken = self.allure_report.broken
        self.passed_rate = self.allure_report.passed_rate
        self.failed_rate = self.allure_report.failed_rate
        self.run_time = self.allure_report.run_time
        self.current_env = os.environ['TEST_ENV']

        self.tester = tester
        self.title = title
        self.report_address = report_address

    def send_text(self, content, mentioned_mobile_list=None):
        """
        发送文本类型通知
        :param content: 文本内容，最长不超过2048个字节，必须是utf8编码
        :param mentioned_mobile_list: 手机号列表，提醒手机号对应的群成员(@某个成员)，@all表示提醒所有人
        :return:
        """
        _DATA = {
            "msgtype": "text",
            "text": {
                "content": content,
                "mentioned_list": None,
                "mentioned_mobile_list": mentioned_mobile_list,
            },
        }

        if mentioned_mobile_list is None or isinstance(mentioned_mobile_list, list):
            # 判断手机号码列表中得数据类型，如果为int类型，发送得消息会乱码
            if len(mentioned_mobile_list) >= 1:
                for i in mentioned_mobile_list:
                    if isinstance(i, str):
                        res = request(
                            url=self.curl,
                            method="post",
                            json=_DATA,
                            headers=self.headers,
                        )
                        if res.json()["errcode"] != 0:
                            raise ValueError(f"企业微信「文本类型」消息发送失败")

                    else:
                        raise TypeError("手机号码必须是字符串类型.")
        else:
            raise ValueError("手机号码列表必须是list类型.")

    def send_markdown(self, content):
        """
        发送 MarkDown 类型消息
        :param content: 消息内容，markdown形式
        :return:
        """
        _DATA = {"msgtype": "markdown", "markdown": {"content": content}}
        res = request(url=self.curl, method="post", json=_DATA, headers=self.headers)
        if res.json()["errcode"] != 0:
            raise ValueError(f"企业微信「MarkDown类型」消息发送失败")

    def articles(self, article):
        """

        发送图文消息
        :param article: 传参示例：{
               "title" : ”标题，不超过128个字节，超过会自动截断“,
               "description" : "描述，不超过512个字节，超过会自动截断",
               "url" : "点击后跳转的链接",
               "picurl" : "图文消息的图片链接，支持JPG、PNG格式，较好的效果为大图 1068*455，小图150*150。"
           }
        如果多组内容，则对象之间逗号隔开传递
        :return:
        """
        _data = {"msgtype": "news", "news": {"articles": [article]}}
        if isinstance(article, dict):
            lists = ["description", "title", "url", "picurl"]
            for i in lists:
                # 判断所有参数都存在
                if article.__contains__(i):
                    res = request(
                        url=self.curl, method="post", headers=self.headers, json=_data
                    )
                    if res.json()["errcode"] != 0:
                        raise ValueError(f"企业微信「图文类型」消息发送失败")
                else:
                    raise ValueError("发送图文消息失败，标题、描述、链接地址、图片地址均不能为空！")
        else:
            raise TypeError("图文类型的参数必须是字典类型")

    def _upload_file(self, file):
        """
        先将文件上传到临时媒体库
        """
        key = self.curl.split("key=")[1]
        url = f"https://qyapi.weixin.qq.com/cgi-bin/webhook/upload_media?key={key}&type=file"
        data = {"file": open(file, "rb")}
        res = request(url, method="post", files=data).json()
        return res["media_id"]

    def send_file_msg(self, file):
        """
        发送文件类型的消息
        @return:
        """

        _data = {"msgtype": "file", "file": {"media_id": self._upload_file(file)}}
        res = request(url=self.curl, method="post", json=_data, headers=self.headers)
        if res.json()["errcode"] != 0:
            raise ValueError(f"企业微信「file类型」消息发送失败")

    def send_msg(self):
        """发送企业微信通知"""
        text = f"""【{self.title}】
                                   >测试环境：<font color=\"info\">{self.current_env}</font>
                                    >测试负责人：{self.tester}
                                    >
                                    > **执行结果**
                                    ><font color=\"info\">🎯运行成功率: {self.passed_rate}</font>
                                    >❤用例  总数：<font color=\"info\">{self.total}个</font>
                                    >😁成功用例数：<font color=\"info\">{self.passed}个</font>
                                    >😭失败用例数：`{self.failed}个`
                                    >😡阻塞用例数：`{self.broken}个`
                                    >😶跳过用例数：<font color=\"warning\">{self.skipped}个</font>
                                    >🕓用例执行时长：<font color=\"warning\">{format_timer(self.run_time)}</font>
                                    >
                                    >测试报告，点击[查看>>测试报告]({self.report_address})"""

        self.send_markdown(text)

    def send_detail_msg(self):
        """
        通知中可根据标记分类显示通过率
        @return:
        """
        text = f"""【{self.title}】
                                   >测试环境：<font color=\"info\">{self.current_env}</font>
                                   >测试负责人：{self.tester}
                                   >
                                   > **执行结果**
                                   ><font color=\"info\">🎯运行成功率: {self.passed_rate}</font>
                                   ><font color=\"red\">🎯运行失败率: {self.failed_rate}</font>
                                   >❤用例  总数：<font color=\"info\">{self.total}个</font>
                                   >😁成功用例数：<font color=\"info\">{self.passed}个</font>
                                   >😭失败用例数：`{self.failed}个`
                                   >😡阻塞用例数：`{self.broken}个`
                                   >😶跳过用例数：<font color=\"warning\">{self.skipped}个</font>
                                   >🕓用例执行时长：<font color=\"warning\">{format_timer(self.run_time)}</font>
                                   """
        if self.report_address:
            text += f""">测试报告，点击[查看>>测试报告]({self.report_address})"""
        self.send_markdown(text)